/*
 * Copyright (c) 2017, 1DAOYUN and/or its affiliates. All rights reserved.
 * 1DAOYUN PROPRIETARY/CONFIDENTIAL. Use is subject to license terms.
 */
package com.iot.nev.gateway.router.kafka;

import java.util.Properties;
import java.util.concurrent.ExecutionException;

import org.apache.kafka.clients.consumer.ConsumerConfig;
import org.apache.kafka.clients.producer.KafkaProducer;
import org.apache.kafka.clients.producer.ProducerConfig;
import org.apache.kafka.clients.producer.ProducerRecord;
import org.apache.log4j.Logger;

import com.iot.nev.gateway.router.Router;
import com.iot.nev.gateway.server.GatewayServer;
import com.iot.nev.gateway.utils.UtilTools;

/**
 * 只负责派发消息，Kafka只起到一个消息服务的（一对多）广播。
 * 
 * 
 * @author Song Xueyong
 *
 */
// TODO: Kafka有Connect开源项目、有MQTT的plugin，
// 目前只完成性能测试，暂时不引入第三方框架。
public class KafkaProducerRouter implements Router {
	/** producer object */
	private KafkaProducer<Integer, String> producer;
	/** logger */
	private Logger logger = Logger.getLogger(KafkaProducerRouter.class);
	/** config */
	private Properties config = UtilTools.getConfig(System.getProperty("user.dir") + "/configuration/config.properties");
	/** 异步派发 */
	private boolean isAsync;
	/** 主题，目前为了模拟性能，主体根据消息头进行划分 */
	private String topic = "Tina-topic";
	/** 分区，目前为了模拟性能，只放一个分区 */
	private int partition = 1;
	/**
	 * Message count as NO. id.
	 */
	private int messageNo = 0;

	/**
	 * 构造函数
	 */
	public KafkaProducerRouter() {
		init();
	}
	
	/**
	 * 每个一个ID.
	 * 
	 * @return
	 */
	public String getID() {
		return KafkaProducerRouter.class.getName() + System.currentTimeMillis();
	}

	/**
	 * 初始化.
	 */
	public void init() {
		Properties props = new Properties();
		props.put(ConsumerConfig.BOOTSTRAP_SERVERS_CONFIG,
				config.getProperty("router_kafka_server_url") + ":" + config.getProperty("router_kafka_server_port"));
		props.put(ProducerConfig.CLIENT_ID_CONFIG, getID());
		props.put(ProducerConfig.KEY_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.IntegerSerializer");
		props.put(ProducerConfig.VALUE_SERIALIZER_CLASS_CONFIG, "org.apache.kafka.common.serialization.StringSerializer");
		producer = new KafkaProducer<>(props);
		
		isAsync = Boolean.valueOf(config.getProperty("router_kafka_server_producer_async"));
		topic = config.getProperty("router_kafka_server_topic");
	}

	public GatewayServer getGateway() {
		return null;
	}

	@Override
	public void publish(String message) {
		try {
			messageNo++;
			// Send asynchronously
			long timestamp = System.currentTimeMillis();
			ProducerRecord<Integer,String> record = new ProducerRecord<>(topic, messageNo, message);
			if (isAsync) {
				producer.send(record, new KafkaProducerCallback(timestamp, messageNo, message));
			} else { // Send synchronously
				// why get object
				Object rmd = producer.send(record).get();
			}
		} catch (InterruptedException | ExecutionException e) {
			e.printStackTrace();
		}
	}
	
	public static void main(String[] args) {
		KafkaProducerRouter kafka = new KafkaProducerRouter();
		int i = 1;
		long s = System.currentTimeMillis();
		do{
			try {
				kafka.publish(i++ + "msg");
				Thread.currentThread().sleep(10);
			} catch (InterruptedException e) {
				e.printStackTrace();
			}
		}while(i<100000);
		System.out.println(System.currentTimeMillis()-s);
	}
}
